package pl.edu.agh.broker.producer;

import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.Map;
import java.util.Map.Entry;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;

import pl.edu.agh.broker.mns.client.DefaultResponder;
import pl.edu.agh.broker.mns.shared.MultiNameService;
import pl.edu.agh.stock.data.IndexDataPoint;
import pl.edu.agh.stock.parser.StockParser;
import pl.edu.agh.stock.parser.exceptions.StockParseException;
import pl.edu.agh.stock.parser.onet.OnetStockParser;

public class StockProducer extends Thread {

	private static final Logger LOGGER = Logger.getLogger(StockProducer.class);

	private final long sleepInterval;
	private final MultiNameService mns;

	private TopicConnection connection;
	private TopicSession session;
	private StockParser stockParser;

	private ArrayList<String> allIndexes = new ArrayList<String>();

	public StockProducer(String brokerURL, long sleepInterval,
			MultiNameService mns) throws JMSException {

		this.sleepInterval = sleepInterval;
		this.mns = mns;

		// create connection factory
		ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
				brokerURL);
		connection = activeMQConnectionFactory.createTopicConnection();
		connection.start();

		// create session, destination and message producer
		session = connection
				.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

		// create stock parser
		stockParser = new OnetStockParser();
	}

	@Override
	public void run() {
		while (!Thread.interrupted()) {
			try {
				Map<String, IndexDataPoint> stockData = stockParser
						.downloadLatestData();
				ArrayList<String> newIndexes = new ArrayList<String>();

				for (Entry<String, IndexDataPoint> indexEntry : stockData
						.entrySet()) {
					newIndexes.add(indexEntry.getValue().getName());
					Message message = session.createObjectMessage(indexEntry
							.getValue());
					Topic topic = session.createTopic(indexEntry.getKey());
					TopicPublisher publisher = session.createPublisher(topic);
					publisher.publish(message);
				}

				if (newIndexes.size() != allIndexes.size()) {
					try {
						allIndexes = newIndexes;
						mns.add("all_indexes", allIndexes,
								new DefaultResponder());
					} catch (RemoteException e) {
						LOGGER.error("Can not register indexes", e);
					}
				}

			} catch (StockParseException e) {
				LOGGER.warn("Can not download latest stock data", e);
			} catch (JMSException e) {
				LOGGER.error("Can not send message", e);
				return;
			}

			try {
				sleep(sleepInterval);
			} catch (InterruptedException e) {
				LOGGER.debug(e);
			}

		}
		try {
			connection.close();
		} catch (JMSException e) {
			LOGGER.info("Can not close connection");
		}
	}
}